use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use anyhow::anyhow;
use chrono::Local;
use fudata::db;
use fudata::db::todo;

use fudata::model::security::SecurityList;
use fudata::model::todo::Todo;
use fudata::model::todo::TodoParameter;
use fudata::model::todo::TodoStatus;
use fudata::model::todo::TodoType;
use log::error;
use log::info;
use tokio::time::Instant;
use zenoh::query::QueryTarget;
use zenoh::selector::Selector;
use zenoh::prelude::r#async::AsyncResolve;

use crate::helper;

use super::update_todo_with_result;

///
/// every day we update the kzz list after 15:30
///
pub async fn generate_todo_sync_securities() -> Result<()> {
    let start = Local::now();
    info!("generate_todo_sync_securities start...");
    let todo_id = Todo::id_from(&[TodoType::SyncSecurities.to_string().as_str()]);

    let paras = get_todo_sync_securities_parameters().await?;
    println!("(generate_todo_sync_securities) parameters={:?}", paras);

    let todo_rs = todo::find(todo_id.as_str()).await;
    match &todo_rs {
        Ok(v) =>
            match v {
                Some(todo) => {
                    // if there is pending todo, then let it be or else update the plan_at to next 1 midnight
                    if todo.is_done() || todo.is_failed() {
                        let plan_at = helper::now_plus_date_time(0, 0, 0, 0);
                        let todo_update = Todo {
                            plan_at: Some(plan_at),
                            status: Some(TodoStatus::Pending),
                            parameters: Some(paras),
                            ..Default::default()
                        };
                        let _ = todo::upsert(todo_id.as_str(), &todo_update).await?;
                    }
                }
                None => {
                    // create a new todo, today midnight
                    let plan_at = helper::now_plus_date_time(0, 0, 0, 0);
                    let todo_new = Todo {
                        name: Some("同步证券列表".to_string()),
                        type_: Some(TodoType::SyncSecurities),
                        // updatable fields
                        plan_at: Some(plan_at),
                        status: Some(TodoStatus::Pending),
                        parameters: Some(paras),
                        created_at: Some(Local::now()),
                        ..Default::default()
                    };
                    let _ = todo::upsert(todo_id.as_str(), &todo_new).await?;
                }
            }
        Err(e) => error!("{}", e),
    }
    info!(
        "generate_todo_sync_securities end/ spent seconds={}",
        Local::now().signed_duration_since(start).num_seconds()
    );

    Ok(())
}

async fn get_todo_sync_securities_parameters() -> Result<Vec<TodoParameter>> {
    let last_updated = db::security::find_max_last_updated().await?;

    Ok(
        vec![TodoParameter {
            name: "last_updated".to_string(),
            value: last_updated,
        }]
    )
}

pub async fn sync_securities(session: Arc<zenoh::Session>, worker_no: u8, todo: &Todo) {
    let max_last_updated = db::security::find_max_last_updated().await;
    let mut rs: Result<()> = Ok(());

    match &max_last_updated {
        Ok(max_last_updated_v) => {
            let selector = Selector::try_from(
                format!("securities?last_updated={}", max_last_updated_v)
            ).unwrap();

            let value = Some(session.zid().to_string());
            let target = QueryTarget::All;
            let timeout = Duration::from_secs(10);

            info!("\n\nSending Query '{selector}'...\n\n");
            let replies_rs = (
                match value {
                    Some(value) => session.get(&selector).with_value(value),
                    None => session.get(&selector),
                }
            )
                .target(target)
                .timeout(timeout)
                .res().await;

            match replies_rs {
                Ok(replies) => {
                    let start = Instant::now();
                    while let Ok(reply) = replies.recv_async().await {
                        match reply.sample {
                            Ok(sample) => {
                                info!(
                                    "\n\n Worker[{worker_no}] Received ('{}': '{}')",
                                    sample.key_expr.as_str(),
                                    sample.value
                                );
                                let ss = SecurityList::try_from(&sample.value);
                                match ss {
                                    Ok(sl) => {
                                        for (index, security) in sl.0.iter().enumerate() {
                                            info!(
                                                "Worker[{worker_no}] Security[{}], : {:?}",
                                                index,
                                                security
                                            );
                                            let _rs = db::security::upsert(&security).await;
                                        }
                                    }
                                    Err(e) => {
                                        error!(
                                            "Worker[{worker_no}] SecurityList::try_from(sample.value){:?}",
                                            e
                                        );
                                        rs = Err(anyhow!(e));
                                        break;
                                    }
                                }
                            }
                            Err(e) => {
                                error!("\n\n Worker[{worker_no}] Received (ERROR: '{}')", e);
                                rs = Err(anyhow!("Received Error: {:?}", e));
                                break;
                            }
                        }
                    }
                    info!(
                        "\n\n Worker[{worker_no}] The query has spent {} seconds",
                        start.elapsed().as_secs()
                    );
                }
                Err(e) => {
                    rs = Err(anyhow!("{}", e));
                }
            }
        }
        Err(e) => {
            error!("(sync::sync_securities) Worker[{worker_no}] {:?}", &max_last_updated);
            rs = Err(anyhow!("{}", e));
        }
    }

    let _ = update_todo_with_result(worker_no, rs, todo).await;
}

#[cfg(test)]
mod test {
    use fudata::db;
    use fudata::model::todo::TodoType;
    use fudata::model::todo::Todo;

    use crate::todo::sync_securities_todo;

    #[tokio::test]
    async fn test_generate_todo_crawl_kzz_list() {
        pretty_env_logger::init_timed();
        let _ = fudata::db::init2("127.0.0.1", 8800).await.unwrap();

        let _ = sync_securities_todo::generate_todo_sync_securities().await;

        let todo_id = Todo::id_from(&[TodoType::SyncSecurities.to_string().as_str()]);

        let rs = db::todo::find(todo_id.as_str()).await;
        println!("rs: {:?}", rs);
    }
}
